/*
Copyright 2024 - 2025 Zen HuiFer

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/

package main

import (
	"context"
	"encoding/json"
	"flag"
	"net/http"
	_ "net/http/pprof"
	"os"
	"strconv"
	"time"

	"github.com/prometheus/client_golang/prometheus/promhttp"
	"go.uber.org/zap"
	"gopkg.in/yaml.v3"
)

var globalConfig ServerConfig

func main() {
	InitLog()

	var configPath string
	flag.StringVar(&configPath, "config", "app-local.yml", "Path to the config file")
	flag.Parse()

	yfile, err := os.ReadFile(configPath)
	if err != nil {
		zap.S().Fatalf("error: %v", err)
	}

	err = yaml.Unmarshal(yfile, &globalConfig)
	if err != nil {
		zap.S().Fatalf("error: %v", err)
	}

	zap.S().Infof("node name = %v , host = %v , port = %v", globalConfig.NodeInfo.Name, globalConfig.NodeInfo.Host, globalConfig.NodeInfo.Port)
	InitRabbitCon()

	initGlobalRedisClient(globalConfig.RedisConfig)

	beforeStart()
	startHttp()

}

func beforeStart() {
	removeOldData()
	go BeatTask(globalConfig.NodeInfo)
	go ListenerBeat()
	go CBeat()
	go timerNoHandlerConfig()


}


var PUSH_CHAN = make(chan []byte, 1000)

func removeOldData() {
	zap.S().Infof("开始清理过期数据")
	HandlerOffNode(globalConfig.NodeInfo.Name)
}

// CBeat 是一个无限循环函数，用于定时检查心跳并进行处理
func CBeat() {
	ticker := time.NewTicker(1 * time.Second)

	for range ticker.C {
		lock := NewRedisDistLock(globalRedisClient, "c_beat")
		if lock.TryLock() {
			service, err := GetThisTypeService()
			if err == nil {
				processHeartbeats(service)
			}
			lock.Unlock()

		} else {

			zap.S().Error("没有获取到 c_beat 处理的锁")

		}

	}
}

// processHeartbeats 函数用于处理心跳信息
//
// 参数：
//
//	service []NodeInfo - 节点信息切片，包含待处理的心跳信息
func processHeartbeats(service []NodeInfo) {
	for _, info := range service {
		if !SendBeat(&info, "beat") {
			globalRedisClient.HDel(context.Background(), "register:"+globalConfig.NodeInfo.Type, info.Name)
			HandlerOffNode(info.Name)
		}

	}
}

// HandlerOffNode 函数用于处理节点下线的情况
//
// 参数：
//
//   - node_name string - 节点名称
//
// 返回值：
//
//   - bool - 如果处理成功返回false，否则返回true
func HandlerOffNode(nodeName string) {
	zap.S().Infof("开始处理节点下线情况, nodeName = %v", nodeName)
	// 清除节点负载计数器

	// 获取节点对应的MQTT客户端ID
	mqttClientIds := GetBindClientId(nodeName)
	zap.S().Infof("获取到节点绑定MQTT客户端ID, nodeName = %v, mqttClientIds = %v", nodeName, mqttClientIds)
	for _, ele := range mqttClientIds {
		zap.S().Infof("获取到节点绑定MQTT客户端ID, nodeName = %v, mqttClientId = %v", nodeName, ele)
		// 获取MQTT客户端ID对应的MQTT配置
		cf := GetUseConfig(ele)
		zap.S().Infof("获取到MQTT客户端ID对应的MQTT配置, client id = %v, config = %v", ele, cf)
		if cf == "" {
			zap.S().Errorf("HandlerOffNode Error get mqtt config, client id = %s", ele)
			continue

		} else {

			var config MqttConfig
			bytes := []byte(cf)
			err := json.Unmarshal(bytes, &config)
			zap.S().Infof("HandlerOffNode 解析MQTT配置成功, config = %v", config)
			if err != nil {
				zap.S().Errorf("HandlerOffNode Error unmarshalling JSON: %s", err)
				continue
			}
			// 移除节点和MQTT客户端ID的关系
			RemoveBindNode(config.ClientId, nodeName)
		}

	}

	CheckMqttConfigIsUsingAndMove(nodeName)
	globalRedisClient.Del(context.Background(), "node_bind:"+nodeName)
	RunCheckMqttConfigIsUsingAndMove = true
}

var RunCheckMqttConfigIsUsingAndMove = false

func startHttp() {
	http.HandleFunc("/beat", HttpBeat)
	http.HandleFunc("/create_mqtt", CreateMqttClientHttp)
	http.HandleFunc("/node_list", NodeList)
	http.HandleFunc("/node_using_status", NodeUsingStatus)
	http.HandleFunc("/mqtt_config", GetUseMqttConfig)
	http.HandleFunc("/no_mqtt_config", GetNoUseMqttConfig)
	http.HandleFunc("/remove_mqtt_client", RemoveMqttClient)
	http.HandleFunc("/push_data", PushMqttData)
	//http.HandleFunc("/pprof-test", Handler)
	http.Handle("/metrics", promhttp.Handler())

	http.HandleFunc("/public_create_mqtt", PubCreateMqttClientHttp)
	http.HandleFunc("/public_remove_mqtt_client", PubRemoveMqttClient)
	http.HandleFunc("/public_push_data", PubPushMqttData)

	if err := http.ListenAndServe(":"+strconv.Itoa(globalConfig.NodeInfo.Port), nil); err != nil {
		zap.S().Fatalf("Failed to start server: %s", err)
	}
}

// timerNoHandlerConfig 函数定时执行 noHandlerConfig 函数
func timerNoHandlerConfig() {
	ticker := time.NewTicker(1 * time.Second)

	if RunCheckMqttConfigIsUsingAndMove {

		for range ticker.C {
			noHandlerConfig()
		}
	}
}

// noHandlerConfig 函数用于处理没有MQTT客户端处理配置的情况
//
// 函数首先创建一个Redis分布式锁，锁的名字为 "no_handler_config_lock"
// 如果成功获取到锁，则打印日志 "获取处理 no_handler_config 的锁"
// 随后调用 GetNoUseConfig 函数获取未使用的MQTT客户端配置列表
// 遍历配置列表，对每一个配置调用 PubCreateMqttClientOp 函数创建MQTT客户端
// 如果创建失败，则继续处理下一个配置
// 最后释放锁资源
//
// 如果获取锁失败，则打印错误日志 "没有获取到 no_handler_config 处理的锁"
func noHandlerConfig() {
	lock := NewRedisDistLock(globalRedisClient, "no_handler_config_lock")
	if lock.TryLock() {

		zap.S().Debugf("获取处理 no_handler_config 的锁")
		configList := GetNoUseConfig()

		for _, conf := range configList {

			if PubCreateMqttClientOp(conf) == -1 {
				continue
			}
			time.Sleep(100 * time.Millisecond)
		}
		lock.Unlock()
	} else {
		zap.S().Error("没有获取到 no_handler_config 处理的锁")
	}

}

// PubCreateMqttClientOp 函数用于创建MQTT客户端
//
// 参数：
//
//   - conf string - MQTT客户端配置信息
//
// 返回值：
//
//   - int - 创建MQTT客户端的结果，成功返回1，失败返回-1
func PubCreateMqttClientOp(conf string) int {
	lose := GetSizeLose("")

	if lose != nil {

		if SendCreateMqttMessage(lose, conf) {
			return 1
		} else {
			zap.S().Errorf("发送 MQTT 客户端创建请求异常 %v", lose)
			return -1
		}

	} else {
		zap.S().Error("没有找到可用节点")
		return -1
	}
}
